/*
* Author: Sowmya Kameswaran
*/
package com.foo.TextDataSender;

import java.util.Properties;

import javax.ejb.Stateless;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.naming.Context;
import javax.naming.InitialContext;

/**
 * Session Bean implementation class DataSender
 */
@Stateless(name="DataSender")
public class DataSender implements DataSenderRemote, DataSenderLocal 
{
		Topic topic = null;
		TopicConnectionFactory factory = null;
		TopicConnection connection = null;
		MessageProducer producer = null;
		MessageProducer snapshotproducer = null;
		MessageProducer temporalproducer = null;
		MessageProducer analyzedproducer = null;
		MessageProducer realtimeproducer = null;
		MessageProducer historicproducer = null;
		TopicSession session = null;
		Topic snapshottopic =null; 
		Topic temporaltopic = null;
		Topic analyzedtopic = null;
		Topic realtimetopic =null;
		Topic historictopic =null;
		
		Queue queue = null;
		QueueConnectionFactory qfactory = null;
		QueueConnection qconnection = null;
		QueueSender qsender = null;
		QueueSession qsession = null;

		@Override
		public void sendMessage(String operation,String data) //TickerInfoData[] data
		{
			try 
			{
				if(operation.equalsIgnoreCase("LoadHistoricalData"))
				{
					//Load Historical Data
					QueueConnectionFactory qfactory = null;
					Properties props = new Properties();
					props.put(Context.PROVIDER_URL, "jnp://localhost:1099");
					props.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory");
					props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming:org.jnp.interfaces");
					Context ctx = new InitialContext(props);
					qfactory = (QueueConnectionFactory) ctx.lookup("XAConnectionFactory");
					qconnection = qfactory.createQueueConnection();
					qsession = qconnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
					queue = (Queue) ctx.lookup("queue/HistoricalDataStorage");
					qsender = qsession.createSender(queue);
					//create and set a message to send
					TextMessage message = qsession.createTextMessage(data);
					qsender.send(message);
					qsession.close ();
					qconnection.close();
				}
				else //Topics
				{
					Properties props = new Properties();
					props.put(Context.PROVIDER_URL, "jnp://localhost:1099");
					props.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory");
					props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming:org.jnp.interfaces");
					Context ctx = new InitialContext(props);
					factory = (TopicConnectionFactory) ctx.lookup("XAConnectionFactory");
					connection = factory.createTopicConnection();
					
					session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
					
					/*qfactory = (QueueConnectionFactory) ctx.lookup("XAConnectionFactory");
					qconnection = qfactory.createQueueConnection();
					qsession = qconnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);*/
					
					if(operation.equals("Spatial"))
					{
						topic = (Topic) ctx.lookup("topic/MySpatialTopic");
						producer = session.createProducer(topic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						producer.send(message);
						System.out.println("Sending message");
						session.close();
						connection.close();
					}
					else if (operation.equals("Snapshot"))
					{
						snapshottopic = (Topic) ctx.lookup("topic/MySnapshotTopic");
						snapshotproducer = session.createProducer(snapshottopic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						//System.out.println("Created text message to send "+message);	
						snapshotproducer.send(message);	
						System.out.println("Sending message");
						session.close();
						connection.close();
					}
					else if(operation.equals("Temporal"))
					{
						temporaltopic = (Topic) ctx.lookup("topic/MyTemporalTopic");
						temporalproducer = session.createProducer(temporaltopic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						//System.out.println("Created text message to send "+message);	
						temporalproducer.send(message);	
						System.out.println("Sending message");
						session.close();
						connection.close();
					}
					else if(operation.equals("Analyzed"))
					{
						analyzedtopic = (Topic) ctx.lookup("topic/MyAnalyzedTopic");
						analyzedproducer = session.createProducer(analyzedtopic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						//System.out.println("Created text message to send "+message);	
						analyzedproducer.send(message);	
						System.out.println("Sending message");
						session.close();
						connection.close();
					}
					else if (operation.equals("RealTime"))
					{
						realtimetopic = (Topic) ctx.lookup("topic/MyRealTimeTopic");
						realtimeproducer = session.createProducer(realtimetopic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						//System.out.println("Created text message to send "+message);	
						realtimeproducer.send(message);	
						System.out.println("Sending message");
						session.close();
						connection.close();
						
					}
					else if (operation.equals("HistoricData"))
					{
						historictopic = (Topic) ctx.lookup("topic/MyHistoricData");
						historicproducer = session.createProducer(historictopic);
						TextMessage message = session.createTextMessage();
						message.setText(data);
						//System.out.println("Created text message to send "+message);	
						historicproducer.send(message);	
						System.out.println("Sending message");
						session.close();
						connection.close();						
					}
				}
			} 
			catch (Exception e) {
				e.printStackTrace();
			}
		}
		/*public static void main(String... args)
		{
			DataSender client = new DataSender();
		}*/
	}

